package com.nami.message.controller.middleware.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.ProtocolStringList;
import com.nami.gate.external.basic.constants.StoreTopicConst;
import com.nami.im.network.model.message.v1.MessageV1;
import com.nami.im.network.model.message.v2.Message;
import com.nami.message.facade.grpmsg.GrpMsgAddFacade;
import com.nami.message.facade.grpmsg.request.GrpMsgAddCmd;
import com.nami.transport.proto.codec.MqTransMsg;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.dromara.common.core.utils.ConverUtils;
import org.dromara.common.json.utils.JsonUtils;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.Set;
import java.util.stream.Collectors;

@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = StoreTopicConst.GCHAT_STORE_MSG_TOPIC,
    consumerGroup = StoreTopicConst.GCHAT_STORE_MSG_CONSUMER_GROUP,
    consumeMode = ConsumeMode.CONCURRENTLY)
public class GrpMsgConsumer implements RocketMQListener<MessageExt> {

    private final GrpMsgAddFacade grpMsgAddFacade;

    @Override
    public void onMessage(MessageExt msg) {
        TimeInterval timer = DateUtil.timer();

        GrpMsgAddCmd grpCmd = parseGrpMsg(msg.getBody());
        grpMsgAddFacade.execute(grpCmd);
        if (timer.interval() > 200) {
            log.warn("[message] clientMsgId={}, msgId={} 群聊写入消耗：{}ms ", grpCmd.getClientUid(),
                grpCmd.getMessageId(), timer.interval());
        }

    }

    private GrpMsgAddCmd parseGrpMsg(byte[] msg) {
        MqTransMsg.GrpLogic2DispatchBody body = null;
        try {
            body = MqTransMsg.GrpLogic2DispatchBody.parseFrom(msg);
        } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException(e);
        } catch (Exception var3) {

        }
        MqTransMsg.TransUser sender = body.getSender();

        Message downMsg = JsonUtils.parseObject(body.getMessage().toStringUtf8(), Message.class);

        GrpMsgAddCmd grpCmd = new GrpMsgAddCmd();
        grpCmd.setPayload(downMsg.getMsgContent());

        grpCmd.setCid(ConverUtils.generateCid(downMsg.getSenderId(), downMsg.getTargetId(), Boolean.TRUE));
        grpCmd.setAppKey(sender.getAppKey());
        grpCmd.setSeq(downMsg.getMsgSeqNo());
        Set<String> ids = body.getReceiverIdsList().stream().collect(Collectors.toSet());
        grpCmd.setReceiverIds(ids);
        grpCmd.setSenderId(downMsg.getSenderId());
        grpCmd.setMessageId(downMsg.getMsgId());
        grpCmd.setSendTime(new Date(downMsg.getMsgTime()));
        grpCmd.setPayloadType(downMsg.getMsgType());
        grpCmd.setReceiverCount(body.getReceiverIdsCount());
        grpCmd.setClientUid(downMsg.getClientUid());

        return grpCmd;
    }

}
